
package org.sopt.dev.ImgKnight.agent.Filter;

import java.util.Collection;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.Vector;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import org.sopt.dev.ImgKnight.agent.database.SeenImageDAO;

public class ImgDupFilter {

	private ConcurrentHashMap<Long, Set<Long>> fpSet = null;
	
	private int seenImageEntryHighWaterMark = -1;
	
	private SeenImageDAO seenImageDAO = null;
	
	private AtomicLong flushDiskCount = null;
	
	private RabinHashFunction fp = null;
	
	private AtomicInteger duplicatedImgCount = null;
	
	private boolean seenImageFlushed = false;
	
	public ImgDupFilter(int seenImageEntryHighWaterMark, boolean isCleanStart) {
		fpSet = new ConcurrentHashMap<Long, Set<Long>>();
		this.seenImageEntryHighWaterMark = seenImageEntryHighWaterMark;
		seenImageDAO = new SeenImageDAO(isCleanStart);
		flushDiskCount = new AtomicLong(seenImageDAO.selectSeenImageSize());
		if( flushDiskCount.longValue() > 0 )
			seenImageFlushed = true;
		fp = new RabinHashFunction();
		duplicatedImgCount = new AtomicInteger(0);
	}
	
	public synchronized List<String> list(String host, List<String> list) {
		Long baseHashCode = fp.hash(host);
		List<Long> fpHashCodes = new Vector<Long>(list.size());
		for(int i=0; i<list.size(); i++)
			fpHashCodes.add(fp.hash(list.get(i)));
		List<String> newList = new LinkedList<String>();
		
		List<Long> subFpHashCodes = fpHashCodes;
		if( seenImageFlushed )
			subFpHashCodes = seenImageDAO.selectDiffFpHashCode(baseHashCode, fpHashCodes);
		
		if( fpSet.containsKey(baseHashCode) ) {
			Set<Long> set = fpSet.get(baseHashCode);
			for(int i=0; i<subFpHashCodes.size(); i++) {
				if( !set.contains(subFpHashCodes.get(i)) ) {
					set.add(subFpHashCodes.get(i));
					int index = fpHashCodes.indexOf(subFpHashCodes.get(i));
					newList.add(list.get(index));
				}
			}
		}
		else {
			fpSet.put(baseHashCode, new HashSet<Long>(subFpHashCodes));
			newList = list;
		}
		
		if( getSeenImageCurrentSize() >= seenImageEntryHighWaterMark )
			memToDist();
		
		return newList;
	}
	
	private synchronized void memToDist() {
		Set<Long> set = fpSet.keySet();
		for(Long key : set) {
			if( seenImageDAO.insertFpHashCode(key, fpSet.get(key)) ) {
				flushDiskCount.addAndGet(fpSet.get(key).size());
				fpSet.remove(key);
				seenImageFlushed = true;
			}
		}
	}
	
	public long addDuplicatedImgCount(int delta) {
		return duplicatedImgCount.addAndGet(delta);
	}
	
	public long getDuplicatedImgCount() {
		return duplicatedImgCount.get();
	}
	
	public long getSeenImageFlushedSize() {
		return flushDiskCount.get();
	}
	
	public synchronized int getSeenImageCurrentSize() {
		int totalEntryCount = 0;
		Collection<Set<Long>> collection = fpSet.values();
		for(Set<Long> set : collection)
			totalEntryCount += set.size();
		return totalEntryCount;
	}
	
}
